package org.example.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/19 10:55
 */
public class MySinkFink {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> streamSource =
                environment.socketTextStream("127.0.0.1", 7777);
        streamSource.addSink(new MySink());
        environment.execute();
    }

    public static class MySink extends RichSinkFunction<String> {

        @Override
        public void open(Configuration parameters) throws Exception {
            // 初始化操作
            super.open(parameters);
        }

        @Override
        public void close() throws Exception {
            // 清理资源
            super.close();
        }

        @Override
        public void invoke(String value, Context context) throws Exception {
            // 拿到一条数据做处理
            super.invoke(value, context);
            System.out.println("invoke---> " + value);
        }
    }

}
